package org.example.reactive;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @author zhaorui
 * @Date 2024/6/15  10:02
 * @description:
 */
public class ReactiveStreanDemo2 {

    public static void main(String[] args) {
        //1.创建一个发布者
        SubmissionPublisher publisher = new SubmissionPublisher();

        //2.创建处理器
        ReactiveProcessor processor = new ReactiveProcessor();

        //3.发布者与处理器建立订阅关系
        publisher.subscribe(processor);

        //4.创建最终的订阅者，即真正的订阅者
        Flow.Subscriber<Object> subscriber = new Flow.Subscriber<>() {
            // 保存订阅关系
            Flow.Subscription sub ;

            @Override
            public void onSubscribe(final Flow.Subscription subscription) {
                this.sub = subscription;

                // 第一次需要
                sub.request(1);
                System.out.println( "建立订阅关系");
            }

            @Override
            public void onNext(final Object item) {
                System.out.println("接收到的数据: "+item);

                // 一些业务处理

                // 消费者需要接收多少数据，是消费者通过这种方式告诉发送者，这样就可以避免消费者被压垮
                sub.request(10);
            }

            @Override
            public void onError(final Throwable throwable) {
                System.out.println("发生了错误:"+ throwable.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("数据发送完会触发的方法");
            }
        };

        //5.中间处理器与订阅者建立订阅关系
        processor.subscribe(subscriber);

        //6.发送数据
        for (int i = 0; i <100;i++){
            publisher.submit("hello,it's me:" + i);
        }

        // 由于响应式编程是非阻塞式的，所以主线程已经进行阻塞，这样 订阅者才能打印出日志
        try {
            Thread.currentThread().join();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

}
